/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.util.Arrays;
import java.util.List;

import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.datatype.DCPackableBytes;
import org.sss.mapreduce.datatype.PackableInt;
import org.sss.mapreduce.datatype.PackableJSONObject;
import org.sss.mapreduce.datatype.PackableString;
import org.sss.server.SssServer;

public class ApacheLogMapper extends Mapper {
  private final Logger logger = SssServer.getLogger(ApacheLogMapper.class);
  private static final PackableInt one = new PackableInt(1);
  private static final String[] acceptableCodes = {
    "200",                      // OK
    "201",                      // Created
    "202",                      // Accepted
    "203",                      // Non-Authoritative Information
    "204",                      // No Content
    "205",                      // Reset Content
    "206"                       // Partial Content
  };
  private static final List<String> acceptableCodeList =
    Arrays.asList(acceptableCodes);
  protected void processData(JSONObject json,
                             Output<PackableString, PackableInt> output) throws Exception {
    String code = (String) json.get("code");
    logger.trace("code = {}", code);
    if (acceptableCodeList.contains(code)) {
      PackableString path = new PackableString((String) json.get("path"));
      output.write(path, one);
    }
  }
  public void map(Context context,
                  DCPackableBytes key, PackableJSONObject val,
                  Output<PackableString, PackableInt> output) throws Exception {
    logger.trace("map method");
    processData(val.get(), output);
  }
}
